﻿//========================================================= 
/**@file ProducerWorker.h
 * @brief 基于线程池的消息多路发送器
 * 
 * @date 2018-08-31   16:53:16
 * @author Zhyioo 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_PRODUCERWORKER_H_
#define _LIBZHOUYB_PRODUCERWORKER_H_
//--------------------------------------------------------- 
#include "../../include/Base.h"

#include <DefaultMQProducer.h>
#include <MQMessageListener.h>
using namespace rocketmq;

#include "../../extension/ability/ThreadTask.h"
using zhou_yb::extension::ability::ThreadTask;
//--------------------------------------------------------- 
namespace zhou_yb {
namespace wrapper {
namespace rocketmq_extractor {
//--------------------------------------------------------- 
/**
 * @brief 消息发送结果回调通知函数
 * @date 2018-08-31 18:10
 */
struct IMessageSendListener
{
    virtual void onSend(const SendResult& rlt, MQMessage& msg) = 0;
    virtual void onException(const MQException& e, MQMessage& msg) = 0;

    virtual ~IMessageSendListener() {}
};
/**
 * @brief 消息发送时的回调函数(用于订制发送逻辑)
 * @date 2018-08-31 18:10
 */
struct IProducerSender
{
    virtual SendResult send(MQMessage& msg) = 0;

    virtual ~IProducerSender() {}
};
//--------------------------------------------------------- 
/**
 * @brief 基于线程池的消息多路分发器
 * @date 2018-08-31 18:02
 */
class ProducerWorker : ThreadTask<MQMessage*>
{
protected:
    Ref<IMessageSendListener> _listener;
    Ref<IProducerSender> _sender;

    virtual void OnTask(ParamType& msg)
    {
        try
        {
            SendResult rlt = _sender->send(*msg);
            if(!_listener.IsNull()) _listener->onSend(rlt, *msg);
        }
        catch(MQException& e)
        {
            if(!_listener.IsNull()) _listener->onException(e, *msg);
        }
        delete msg;
    }
    virtual void OnQuit(ParamType& msg)
    {
        delete msg;
    }
public:
    ProducerWorker() : ThreadTask<MQMessage*>() {}
    void Init(Ref<IProducerSender> sender, size_t thread_count,
        Ref<IMessageSendListener> listener = Ref<IMessageSendListener>())
    {
        if(sender.IsNull())
            return;

        _sender = sender;
        _listener = listener;
        ThreadTask<MQMessage*>::Init(thread_count);
    }
    void Push(const string& topic, const string& tag, 
        const string& key, const string& body)
    {
        MQMessage* msg = new MQMessage();
        msg->setTopic(topic);
        msg->setTags(tag);
        if(!key.empty()) msg->setKeys(key);
        msg->setBody(body);

        ThreadTask<MQMessage*>::Push(msg);
    }
};
//--------------------------------------------------------- 
} // namespace rocketmq_extractor 
} // namespace wrapper
} // namespace zhou_yb 
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_PRODUCERWORKER_H_
//========================================================= 